package demo3_5;
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.util.Collection;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Random;
import java.util.Set;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
public class SplitGenres {
	private static long train_male = 0;
	private static long train_female = 0;
	private static long test_male = 0;
	private static long test_female = 0;
	
	private static Set<Integer> male  = new HashSet<Integer>() ; //男性
	private static Set<Integer> female  = new HashSet<Integer>() ; //女性
	
	public static void main(String[] args) throws Exception {
		Configuration conf=new Configuration();
		conf.set("fs.defaultFS", "DragonBoy:9000");
		FileSystem fs=FileSystem.get(conf);
		
		Path moviedata=new Path("/movie/processing_out/part-m-00000");

		int datasize=getSize(fs, moviedata);
		System.out.println("总记录数:"+datasize);
		System.out.println("男总记录为："+male.size());
		System.out.println("女总记录为："+female.size());
		
		
		Path nan=new Path("hdfs://DragonBoy:9000/movie/male");
		fs.delete(nan,true);
		FSDataOutputStream os1=fs.create(nan);
		BufferedWriter bw1=new BufferedWriter(new OutputStreamWriter(os1));
		
		Path nv=new Path("hdfs://DragonBoy:9000/movie/female");
		fs.delete(nv,true);
		FSDataOutputStream os2=fs.create(nv);
		BufferedWriter bw2=new BufferedWriter(new OutputStreamWriter(os2));
		
		Path kong=new Path("hdfs://DragonBoy:9000/movie/nullData");
		fs.delete(kong,true);
		FSDataOutputStream os3=fs.create(kong);
		BufferedWriter bw3=new BufferedWriter(new OutputStreamWriter(os3));
		
//		
//		Path train=new Path("hdfs://DragonBoy:9000/movie/trainData");
//		fs.delete(train,true);
//		FSDataOutputStream os1=fs.create(train);
//		BufferedWriter bw1=new BufferedWriter(new OutputStreamWriter(os1));
//	
//		Path test=new Path("hdfs://DragonBoy:9000/movie/testData");
//		fs.delete(test,true);
//		FSDataOutputStream os2=fs.create(test);
//		BufferedWriter bw2=new BufferedWriter(new OutputStreamWriter(os2));
//
//		Path validate=new Path("hdfs://DragonBoy:9000/movie/validateData");
//		fs.delete(validate,true);
//		FSDataOutputStream os3=fs.create(validate);
//		BufferedWriter bw3=new BufferedWriter(new OutputStreamWriter(os3));
//		
		FSDataInputStream is=fs.open(moviedata);
		BufferedReader br=new BufferedReader(new InputStreamReader(is));
//		
//		
//		Set<Integer> train_index=trainIndex(datasize);
//		
//		
//		Set<Integer> validate_index=validateIndex(datasize,train_index);
		String line="";
//		String subLine="";
//		
		int nullData = 0; //空数据
		int sum=0;
//		int trainsize=0;
//		int testsize=0;
//		int validatesize=0;
//		
		while((line=br.readLine())!=null){
			
			if(male.contains(sum)){
				//1,1,1,10,48067,5,5,18,20,14,2,0,21,3,0,0,14,0,6,3,3,2,0
				bw1.write(line.toString());
				bw1.newLine();
			}else if(female.contains(sum)){
				bw2.write(line.toString());
				bw2.newLine();
			}else{
				nullData+=1;
				bw3.write(line.toString());
				bw3.newLine();
			}
			sum+=1;
		}
		System.out.println("无人认领的空数据为："+nullData);
		
		
//		while((line=br.readLine())!=null){
//			sum+=1;
//			if(train_index.contains(sum)){
//				trainsize+=1;
//				//1,1,1,10,48067,5,5,18,20,14,2,0,21,3,0,0,14,0,6,3,3,2,0
//				subLine = line.substring(line.indexOf(",")+1,line.indexOf(",")+2);
//				
//				if(subLine.equals("0")) {
//					train_male++;
//				}else {
//					train_female++;
//				}
//				
//				bw1.write(line.toString());
//				bw1.newLine();
//			}else if(validate_index.contains(sum)){
//				validatesize+=1;
//				bw3.write(line.toString());
//				bw3.newLine();
//			}else{
//				testsize+=1;
//				subLine = line.substring(line.indexOf(",")+1,line.indexOf(",")+2);
//				if(subLine.equals("0")) {
//					test_male++;
//				}else {
//					test_female++;
//				}
//				
//				bw2.write(line.toString());
//				bw2.newLine();
//			}
//		}
//		System.out.println("训练数据总记录数:"+trainsize);
//		
//		
//		System.out.println("训练里的男数据："+train_male);
//		System.out.println("训练里的女数据："+train_female);
//		
//		System.out.println("测试数据总记录数:"+testsize);
//		
//		System.out.println("测试里的男数据："+test_male);
//		System.out.println("测试里的女数据："+test_female);
		
		bw1.close();
		os1.close();
		bw2.close();
		os2.close();
		bw3.close();
		os3.close();
		br.close();
		is.close();
		fs.close();
	}
	/**
	 * 读取原始数据并统计数据的记录数
	 * @param fs
	 * @param path
	 * @return
	 * @throws Exception
	 */
	public static int getSize(FileSystem fs,Path path) throws Exception{
		int count=0;
		FSDataInputStream is=fs.open(path);
		BufferedReader br=new BufferedReader(new InputStreamReader(is));
		String line="";
		String subLine = "";
		while((line=br.readLine())!=null){
			subLine = line.substring(line.indexOf(",")+1,line.indexOf(",")+2);
			if(subLine.equals("0")) {
				male.add(count);
			}else {
				female.add(count);
			}
			count++;
		}
		br.close();
		is.close();
		return count;		
	}
//	/**
//	 *随机获取80%的原始数据下标 -训练数据
//	 * @param count
//	 * @return
//	 */
//	public static Set<Integer> trainIndex(int count){
//		Set<Integer> train_index=new HashSet<Integer>();
//		int trainSplitNum=(int)(count*0.8);
//		Random random=new Random();
//		while(train_index.size()<trainSplitNum){
//			int a=random.nextInt(count);
//			train_index.add(a);
//		}
//		return train_index;	
//	}
//	/**
//	 * 随机获取10%的原始数据下标-验证数据
//	 * @param count
//	 * @param train_index
//	 * @return
//	 */
//	public static Set<Integer> validateIndex(int count,Set<Integer> train_index){
//		Set<Integer> validate_index=new HashSet<Integer>();
//		int validateSplitNum=count-(int)(count*0.9);
//		Random random=new Random();
//		while(validate_index.size()<validateSplitNum){
//			int a=random.nextInt(count);
//			if(!train_index.contains(a)){
//				validate_index.add(a);	
//			}
//		}
//		return validate_index;	
//	}
	
}
